package com.song.hbaseChap_5_02;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.sun.corba.se.spi.ior.Writeable;

public class HbaseOuputMapreduce extends Configured implements Tool {
	
	

	
	@Override
	public int run(String[] args) throws Exception {
		
		String input =  "hdfs://macsong:9000/input/data";
		String table =  "demoout";
//		String columns = args[2];
		Configuration conf = HBaseConfiguration.create();
		conf.set("hbase.master", "macsong:60000");
		conf.set("hbase.zookeeper.quorum", "macsong");
		
		Job job = new Job(conf , " imple from file "+input +" into table "+table);
		
		job.setJarByClass(HbaseOuputMapreduce.class);
		
		job.setMapperClass(ExampleOutMapper.class );
		job.setOutputFormatClass(TableOutputFormat.class);
		job.getConfiguration().set(TableOutputFormat.OUTPUT_TABLE, table);
		job.setOutputKeyClass(ImmutableBytesWritable.class);
		job.setOutputValueClass(Writable.class);
		job.setNumReduceTasks(0);
		FileInputFormat.addInputPath(job, new Path(input));
		return job.waitForCompletion(true)? 0:1;
	}

	public static void main(String[] args) {
		Configuration conf = new Configuration();
		
		
		try {
			ToolRunner.run(conf, new HbaseOuputMapreduce(),args);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}
	
	
}
